// @Author EthanScriptOn
// @Desc
package extend_param_builder

import (
	"fmt"
	"gitee.com/fatzeng/srf_switch_expansion_package/extend_param_collector"
	"github.com/confluentinc/confluent-kafka-go/kafka"
)

type KafkaProducerParamsBuilder struct {
}

func (k *KafkaProducerParamsBuilder) Build(params interface{}) (interface{}, error) {
	producerParams, ok := params.(*extend_param_collector.KafkaProducerParams)
	if !ok {
		return nil, fmt.Errorf("params is not KafkaProducerParams")
	}
	if !producerParams.IsValid() {
		return nil, fmt.Errorf("KafkaProducerParams is not valid")
	}
	messages := make([]*kafka.Message, 0)
	for _, param := range producerParams.Params {
		headers := make([]kafka.Header, 0)
		for _, header := range param.Headers {
			headers = append(headers, kafka.Header{
				Key:   header.Key,
				Value: header.Value,
			})
		}
		singleMsg := kafka.Message{
			TopicPartition: kafka.TopicPartition{
				Topic:     &param.Topic,
				Partition: param.Partition,
				Offset:    kafka.Offset(param.Offset),
				Metadata:  &param.Metadata,
			},
			Value:         param.Value,
			Key:           param.Key,
			Timestamp:     param.Timestamp,
			TimestampType: kafka.TimestampType(param.TimestampType),
			Headers:       headers,
		}
		messages = append(messages, &singleMsg)
	}
	return messages, nil
}
